package com.demo.common.mq;

import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.demo.common.data.LocalDataPersistent;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 本地消息队列
 */
@Component
@ConditionalOnExpression("#{'local'.equals(environment['moudle-type.mq'])}")
@Slf4j
public class LocalMQ implements MQTemplate , LocalDataPersistent, ApplicationRunner {

    /**
     * 消息处理线程池
     */
    private final ThreadPoolExecutor listenerExecutors = new ThreadPoolExecutor(
            5,
            5,
            30,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(Integer.MAX_VALUE),
            new ThreadFactoryBuilder().setNameFormat("localMQListen-%d").build());

    /**
     * 消费程池
     */
    private final ThreadPoolExecutor consumerExecutors = new ThreadPoolExecutor(
            2,
            2,
            30,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(Integer.MAX_VALUE),
            new ThreadFactoryBuilder().setNameFormat("localConsumer-%d").build());
    @Resource
    private MessageHandler messageHandler;

    private Queue<LocalMessage> messageQueue=new ConcurrentLinkedQueue<>();
    /**
     * 标志位，减少lock的判断
     */
    private volatile boolean running=false;

    private ReentrantLock lock=new ReentrantLock();

    @PreDestroy
    public void saveMemoryQueue(){
        running=false;
        this.saveData(messageQueue);
    }

    public LocalMQ(){
        Queue<LocalMessage> localMessageQueue=this.getSavedData(new TypeReference<LinkedBlockingQueue<LocalMessage>>() {});
        if(localMessageQueue!=null&&!localMessageQueue.isEmpty()){
            messageQueue=localMessageQueue;
        }
    }

    /**
     * 比死循环效率高，不会空跑
     */
    @SneakyThrows
    private void consumerMessage(){
        if(running){
            log.info("当前队列已在消费中");
            return;
        }
        boolean tryLock = lock.tryLock();
        if(tryLock){
            log.info("消费消息开始");
            running=true;
            try{
                while (running){
                    LocalMessage poll = messageQueue.poll();
                    if(poll!=null){
                        listenerExecutors.execute(()->{
                            boolean result=messageHandler.handleMsg(poll.getQueue(), poll.getMsg());
                            if(!result){
                                poll.setHandleCount(poll.getHandleCount()+1);
                                if(poll.getHandleCount()<3){
                                    //重新把消费仍回去，类似于rabbitMq的ack机制
                                    messageQueue.add(poll);
                                }else{
                                    log.warn("队列{}消息{}重试超过上限，消息丢弃",poll.getQueue(),poll.getMsg());
                                }
                            }
                        });
                    }else {
                        running=false;
                        break;
                    }
                }
                log.info("消费消息结束");
            }finally {
                lock.unlock();
            }
        }else{
            log.info("获取锁失败，已有其他线程处理中");
        }
    }

    @Override
    @SneakyThrows
    public boolean sendMsg(String queueName, String msg) {
        messageQueue.add(new LocalMessage(queueName,msg,0));
        //这么处理的好处是不用空跑，效率高
        consumerExecutors.execute(()->consumerMessage());
        return true;
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        if(!messageQueue.isEmpty()&&!running){
            log.info("开始消费反序列化信息");
            consumerMessage();
        }
    }
}
